package com.atguigu.writable2;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/*
  在该类中去实现MapTask中实现的业务逻辑代码
     Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> :
        两组：
            KEYIN：读取数据的偏移量的类型
            VALUEIN：读取的一行一行的数据的类型

             KEYOUT：写出的key的类型（在这是手机号）
             VALUEOUT：写出的value的类型（在这是FlowBean）

      注意：泛型上的类型必须能够序列化
 */
public class FlowMapper extends Mapper<LongWritable, Text,Text,FlowBean> {
    //封装key
    private Text outKey = new Text();
    //封装value
    private FlowBean outValue = new FlowBean();
    /**
     * map方法用来实现MapTask中需要实现的业务逻辑代码
     * map方法在被循环调用，每调用一次传入一行数据
     * @param key 读取数据时的偏移量
     * @param value 一行一行的数据
     * @param context 上下文（在这是用来将key,value写出去）
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
       //1.将Text变成String
        String line = value.toString();
        //2.切割数据
        String[] split = line.split("\t");
        //3.封装Key,Value
        //给key赋值
        outKey.set(split[1]);
        //给value赋值
        outValue.setUpFlow(Long.parseLong(split[split.length -3]));
        outValue.setDownFlow(Long.parseLong(split[split.length -2]));
        outValue.setSumFlow(outValue.getUpFlow() + outValue.getDownFlow());
        //4.将key,value写出去
        context.write(outKey,outValue);
    }
}
